/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.Record;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.paxos2.*;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.txn.TxnHeader;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;

/**
 * This class has the control logic for the Follower.
 */
public class Follower extends Learner {

	private long lastQueued;
	// This is the same object as this.zk, but we cache the downcast op
	final FollowerZooKeeperServer fzk;

	Follower(QuorumPeer self, FollowerZooKeeperServer zk) {
		this.self = self;
		this.zk = zk;
		this.fzk = zk;
	}

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append("Follower ").append(sock);
		sb.append(" lastQueuedZxid:").append(lastQueued);
		sb.append(" pendingRevalidationCount:").append(
				pendingRevalidations.size());
		return sb.toString();
	}

	/**
	 * the main method called by the follower to follow the leader
	 * 
	 * @throws InterruptedException
	 */
	void followLeader() throws InterruptedException {
		try {
			InetSocketAddress addr = findLeader();
			try {
				connectToLeader(addr);
				long newLeaderZxid = registerWithLeader(Leader.FOLLOWERINFO);
				// check to see if the leader zxid is lower than ours
				// this should never happen but is just a safety check
				long lastLoggedZxid = self.getLastLoggedZxid();
				if ((newLeaderZxid >> 32L) < (lastLoggedZxid >> 32L)) {
					LOG.fatal("Leader epoch "
							+ Long.toHexString(newLeaderZxid >> 32L)
							+ " is less than our epoch "
							+ Long.toHexString(lastLoggedZxid >> 32L));
					throw new IOException("Error: Epoch of leader is lower");
				}
				syncWithLeader(newLeaderZxid);
				QuorumPacket qp = new QuorumPacket();
				while (self.isRunning()) {
					readPacket(qp);
					processPacket(qp);
				}
			} catch (IOException e) {
				LOG.warn("Exception when following the leader", e);
				try {
					sock.close();
				} catch (IOException e1) {
					e1.printStackTrace();
				}

				synchronized (pendingRevalidations) {
					// clear pending revalidations
					pendingRevalidations.clear();
					pendingRevalidations.notifyAll();
				}
			}
		} finally {
		}
	}

	/**
	 * Examine the packet received in qp and dispatch based on its contents.
	 * 
	 * @param qp
	 * @throws IOException
	 */
	protected void processPacket(QuorumPacket qp) throws IOException {
		switch (qp.getType()) {
		case Leader.PING:
			ping(qp);
			break;
		case Leader.PROPOSAL:
			BinaryInputArchive ia = BinaryInputArchive
					.getArchive(new ByteArrayInputStream(qp.getData()));
			int size = ia.readInt("size");

			TxnHeader[] hdr = new TxnHeader[size];
			Record[] txn = new Record[size];

			for (int i = 0; i < size; i++) {
				hdr[i] = new TxnHeader();
				txn[i] = SerializeUtils.deserializeTxn(ia, hdr[i]);
			}
			if (hdr[0].getZxid() != lastQueued + 1) {
				LOG.warn("Got zxid 0x" + Long.toHexString(hdr[0].getZxid())
						+ " expected 0x" + Long.toHexString(lastQueued + 1));
			}
			lastQueued = hdr[hdr.length - 1].getZxid();
			fzk.logRequest(hdr, txn);
			break;
		case Leader.COMMIT:
			//System.out.println("commit "+qp.getZxid());
			fzk.commit(qp.getZxid());
			break;
		case Leader.UPTODATE:
			LOG.error("Received an UPTODATE message after Follower started");
			break;
		case Leader.REVALIDATE:
			revalidate(qp);
			break;
		case Leader.SYNC:
			fzk.sync();
			break;
		case Leader.QUERY_LEADER:
			if (Configuration.debug)
				System.out.println("QueryLeaderAck " + qp.getZxid());
			QuorumPacket ack = new QuorumPacket(Leader.QUERY_LEADER_ACK, qp
					.getZxid(), null);
			try {
				this.writePacket(ack, true);
			} catch (IOException e) {
				LOG
						.warn(
								"Closing connection to leader, exception during packet send",
								e);
				try {
					if (!this.sock.isClosed()) {
						this.sock.close();
					}
				} catch (IOException e1) {
					// Nothing to do, we are shutting things down, so an
					// exception here is irrelevant
					LOG.debug("Ignoring error closing the connection", e1);
				}
			}
			break;
		case Leader.BOLOSKY_READ:
			ObjectInputStream ois = new ObjectInputStream(
					new ByteArrayInputStream(qp.getData()));
			size = ois.readInt();
			Request[] reqs = new Request[size];
			for (int i = 0; i < size; i++) {
				long sessionId = ois.readLong();
				int cxid = ois.readInt();
				int nonce = ois.readInt();
				int type = ois.readInt();
				int reqSize = ois.readInt();
				byte[] data = new byte[reqSize];
				ois.readFully(data);
				ByteBuffer buffer = ByteBuffer.wrap(data);
				Request req = new Request(sessionId, cxid, nonce, type, buffer);
				req.zxid = qp.getZxid();
				reqs[i] = req;
				if (Configuration.debug)
					System.out.println("Bolosky read " + sessionId + " " + cxid
							+ " blockId=" + Mapping.getBlockId(req) + " zxid="
							+ qp.getZxid());
			}
			fzk.boloskyRead(reqs);
			break;

		}
	}

	/**
	 * The zxid of the last operation seen
	 * 
	 * @return zxid
	 */
	public long getZxid() {
		try {
			synchronized (fzk) {
				return fzk.getZxid();
			}
		} catch (NullPointerException e) {
			LOG.warn("error getting zxid", e);
		}
		return -1;
	}

	/**
	 * The zxid of the last operation queued
	 * 
	 * @return zxid
	 */
	protected long getLastQueued() {
		return lastQueued;
	}

	@Override
	public void shutdown() {
		LOG.info("shutdown called", new Exception("shutdown Follower"));
		super.shutdown();
	}
}
